package com.atguigu.flink.sqlfunction;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * Created by Smexy on 2023/2/6
 *
 *  UDAF:  和group by结合使用。
 *          多进(N行N列)1出(1行1列)
 *
 *     求每种传感器的平均VC
 */
public class Demo4_AggFunction
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = "create table t1( id string,ts bigint ,vc int ) with(" +
            "                    'connector' = 'filesystem' ," +
            "                    'path' = 'data/func/sensor.txt' , " +
            "                    'format' = 'csv'  " +
            "                    )";

        //执行建表(写操作)
        tableEnvironment.executeSql(createTableSql);

        //把t1变为Table对象
        Table table = tableEnvironment.from("t1");

        //1.先创建一个函数对象
        MyAgg myFunc = new MyAgg();

        //注册函数，再用
        tableEnvironment.createTemporaryFunction("a",myFunc);

        //TableAPI使用
       /* table.groupBy($("id"))
             .select($("id"),call("a",$("vc")).as("avgVc"))
             .execute().print();*/

       tableEnvironment.sqlQuery("select id,a(vc) avgVc from t1 group by id").execute().print();

    }

    /*
            必须实现:
                createAccumulator()
                accumulate()
                getValue()
     */
    public static class MyAgg extends AggregateFunction<Double,MyAcc>{

        public void accumulate(MyAcc myAcc,Integer vc){
            myAcc.num += 1;
            myAcc.vc += vc;

        }
        //返回最终结果
        @Override
        public Double getValue(MyAcc accumulator) {
            return accumulator.vc / accumulator.num;
        }

        //创建累加器
        @Override
        public MyAcc createAccumulator() {
            return new MyAcc(0,0d);
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyAcc{
        private Integer num;
        private Double vc;
    }


}
